package com.briup.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

/**
 * 找出每一年的最高温度
 * 输入处理器使用KeyValueTextInputFormat  告诉输入处理器要处理的文件路径
 * mapper 使用默认的Mapper
 * 设置map阶段的k2v1输出类型 Text  Text
 * partition 一个reducerTask 所以不需要处理分区
 * reducer 要自定义  选出最高温度
 * reduce输出类型 k3 v3 Text  IntWritable
 * 输出处理器 使用默认的  也需要指定输出路径
 */
public class MaxWeather extends Configured implements Tool {
    /**
     * 找出最高温度
     */
    public static  class  MaxWeatherReducer extends Reducer<Text,Text,Text, IntWritable>{

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            System.out.println("key = " + key.toString());
            int  maxWeather=0;
//            遍历values 找出最高温度  基于选择排序的思想
            for (Text value : values) {
//                将序列化类型转成String
                String str = value.toString();
//                要String 转换成int
                int weather = Integer.parseInt(str);
//                与变量"最高温度"比较 每一个温度  如果大于最高温度 当前温度就是最高温度
                if(weather>maxWeather)
                    maxWeather=weather;
            }
            context.write(key,new IntWritable(maxWeather));
        }
    }


    @Override
    public int run(String[] args) throws Exception {
//        获取配置对象
        Configuration conf = getConf();
//        配置对象设置 切割参数
//        设置参数的时候不能使用正则表达
        conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator",",");
//        从配置对象获取外部参数 in和out
        String in = conf.get("in");
        String out = conf.get("out");

//        创建job对象 使用配置对象
        Job job = Job.getInstance(conf);
//        为job对象所设置基本属性  jobName  运行那个jar包
        job.setJobName("maxweather");
        job.setJarByClass(this.getClass());
//        输入处理器要使用KeyValueTextInPutFormat
        job.setInputFormatClass(KeyValueTextInputFormat.class);
//        告诉输入处理器 输入文件的路径（现场创建输入路径对象）
        KeyValueTextInputFormat.addInputPath(job,new Path(in));
//        Mapper使用默认的Mapper 可以不设置
//        map阶段的输出k：Text  v：Text的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
//        分区器 partitioner 不需要设置使用默认的就行  因为reducerTask个数为1  默认也是1个 也不需要设置
//        设置Reducer 要自定义类型
        job.setReducerClass(MaxWeatherReducer.class);
//        设置reduce阶段 k： Text v: IntWritable的输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
//        设置输出处理器  使用默认的就可以  就不需要设置了
//        告诉输出处理器输出路径
        TextOutputFormat.setOutputPath(job,new Path(out));
//        提交job

        return job.waitForCompletion(true)?0:-1;
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new MaxWeather(),args);
    }
}
